package com.sl.core.engine.thread;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.ReflectUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sl.core.engine.trans.ProcessInstance;
import lombok.*;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

@Slf4j
public class TraceThreadPoolTaskExecutor extends ThreadPoolExecutor {

    @Getter
    private String threadNamePrefix;

    public static final Cache<String, List<TrackFutureInfo>> THREAD_CACHE = CacheBuilder.newBuilder().initialCapacity(1000).maximumSize(50000).expireAfterAccess(4, TimeUnit.HOURS).build();


    public TraceThreadPoolTaskExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, Integer queueCapacity, String threadNamePrefix) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new LinkedBlockingQueue(queueCapacity));
        this.threadNamePrefix = threadNamePrefix;
        ThreadFactory build = ThreadFactoryBuilder.create().setNamePrefix(threadNamePrefix).build();
        this.setThreadFactory(build);
    }

    public static Map<String, String> getThreadStackTraceMap(String threadNamePrefix) {
        Map<String, String> threadStackTraceMap = Maps.newHashMap();

        Map<Thread, StackTraceElement[]> allStackTraces = Thread.getAllStackTraces();
        for (Map.Entry<Thread, StackTraceElement[]> threadEntry : allStackTraces.entrySet()) {
            Thread key = threadEntry.getKey();
            StackTraceElement[] value = threadEntry.getValue();
            if (!StrUtil.startWith(key.getName(), threadNamePrefix)) {
                continue;
            }
            threadStackTraceMap.put(key.toString(), getStackMsg(value));
        }
        return threadStackTraceMap;
    }

    @SneakyThrows
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        String trackId = MDC.get(ProcessInstance.TID_IDENTIFY);
        return this.submit(task, trackId);
    }

    public TrackFutureInfo addTrackFuture(Thread thread) {
        String trackId = MDC.get(ProcessInstance.TID_IDENTIFY);
        return this.addTrackFuture(thread, trackId);
    }

    @SneakyThrows

    public TrackFutureInfo addTrackFuture(Thread thread, String tid) {
//        ReflectUtil.setFieldValue(future, "runner", thread);
        if (StrUtil.isBlank(tid)) {
            return null;
        }
        Future future = new TrackFutureTask(thread);
        List<TrackFutureInfo> trackFutureInfos = THREAD_CACHE.get(tid, Lists::newArrayList);
        TrackFutureInfo trackFutureInfo = new TrackFutureInfo(future);
        trackFutureInfos.add(trackFutureInfo);
        return trackFutureInfo;
    }


    public Future<?> submit(Runnable task) {
        String trackId = MDC.get(ProcessInstance.TID_IDENTIFY);
        return this.submit(task, trackId);
    }

    @SneakyThrows
    public Future<?> submit(Runnable task, String tid) {
        this.invalid();

        String instanceId = MDC.get(ProcessInstance.INSTANCE_IDENTIFY);

        Future<?> submit = super.submit(() -> {
            MDC.put(ProcessInstance.TID_IDENTIFY, tid);
            MDC.put(ProcessInstance.INSTANCE_IDENTIFY, instanceId);
            task.run();
        });
        if (StrUtil.isBlank(tid)) {
            return submit;
        }

        List<TrackFutureInfo> trackFutureInfos = THREAD_CACHE.get(tid, Lists::newArrayList);
        TrackFutureInfo trackFutureInfo = new TrackFutureInfo(submit);
        trackFutureInfos.add(trackFutureInfo);
        return submit;
    }


    public void close(String trackId) {
        List<TrackFutureInfo> trackFutureInfos = THREAD_CACHE.getIfPresent(trackId);
        if (CollUtil.isEmpty(trackFutureInfos)) {
            return;
        }
        List<TrackFutureInfo> reverse = CollUtil.reverse(trackFutureInfos);
        for (TrackFutureInfo trackFutureInfo : reverse) {
            Future<?> future = trackFutureInfo.getFuture();
            if (future.isDone() || future.isCancelled()) {
                continue;
            }
            kill(future, true);
        }
        THREAD_CACHE.invalidate(trackId);
    }

    public static void kill(Future<?> future, Boolean forceFlag) {
        if (future instanceof FutureTask && forceFlag) {
            try {
                if (future instanceof TrackFutureTask) {
                    future.cancel(true);
                    return;
                }
                Object runner = ReflectUtil.getFieldValue(future, "runner");
                if (ObjectUtil.isNotNull(runner)) {
                    Thread thread = (Thread) runner;
                    thread.interrupt();
                    if (!thread.isInterrupted()) {
//                        thread.stop();
                    }
                    // }
                    return;
                }
            } catch (Exception e) {
                log.error("停止线程失败 降级使用cancel", e);
            }
        }

        future.cancel(true);
    }


    public ConcurrentMap<String, List<TrackFutureInfo>> getRunnerFuture() {
        this.invalid();
        return THREAD_CACHE.asMap();
    }

    @SneakyThrows
    public void removeTrackFuture(TrackFutureInfo trackFutureInfo) {
        if (ObjectUtil.isNull(trackFutureInfo)) {
            return;
        }
        String trackId = MDC.get(ProcessInstance.TID_IDENTIFY);
        if (StrUtil.isBlank(trackId)) {
            return;
        }
        List<TrackFutureInfo> trackFutureInfos = THREAD_CACHE.get(trackId, Lists::newArrayList);
        trackFutureInfos.remove(trackFutureInfo);
    }


    @SneakyThrows
    public <T> Future<T> submit(Callable<T> task, String tid) {
        this.invalid();
        Future<T> submit = super.submit(task);
        if (StrUtil.isBlank(tid)) {
            return submit;
        }

        List<TrackFutureInfo> trackFutureInfos = THREAD_CACHE.get(tid, Lists::newArrayList);
        TrackFutureInfo trackFutureInfo = new TrackFutureInfo(submit);
        trackFutureInfos.add(trackFutureInfo);
        return submit;
    }

    @SneakyThrows
    public void execute(Thread command) {
        String trackId = MDC.get(ProcessInstance.TID_IDENTIFY);
        this.execute(command, trackId);
    }

    @SneakyThrows
    public void execute(Thread command, String tid) {
        this.invalid();
        super.execute(command);
        if (StrUtil.isBlank(tid)) {
            return;
        }
        List<TrackFutureInfo> trackFutureInfos = THREAD_CACHE.get(tid, Lists::newArrayList);

        Future future = new TrackFutureTask(command);
        TrackFutureInfo trackFutureInfo = new TrackFutureInfo(future);
        trackFutureInfos.add(trackFutureInfo);
    }

    private void invalid() {
        try {
            List<String> validKey = Lists.newArrayList();

            THREAD_CACHE.asMap().forEach((k, fi) -> {
                Iterator<TrackFutureInfo> it_b = fi.iterator();
                while (it_b.hasNext()) {
                    TrackFutureInfo v = it_b.next();
                    if (ObjectUtil.isNull(v)) {
                        it_b.remove();
                        continue;
                    }
                    Future<?> future = v.getFuture();
                    if (future.isDone() || future.isCancelled()) {
                        it_b.remove();
                    }
                }

                if (CollUtil.isEmpty(fi)) {
                    validKey.add(k);
                }
            });
            if (CollUtil.isNotEmpty(validKey)) {
                THREAD_CACHE.invalidateAll(validKey);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static String getStackMsg(StackTraceElement[] stackArray) {

        StringBuffer sb = new StringBuffer();
        for (int i = 0; i < stackArray.length; i++) {
            StackTraceElement element = stackArray[i];
            sb.append(element.toString() + "\n");
        }
        return sb.toString();
    }

    public static class TrackFutureTask<T> extends FutureTask<T> {

        private Thread thread;

        public TrackFutureTask(Thread thread) {
            super(() -> {
                return null;
            });
            this.thread = thread;
        }

        @Override
        public boolean isDone() {
            if (ObjectUtil.isNull(thread)) {
                return true;
            }
            Thread.State state = thread.getState();
            if (state != Thread.State.RUNNABLE) {
                return true;
            }
            return false;
        }

        @Override
        public boolean isCancelled() {
            if (ObjectUtil.isNull(thread)) {
                return true;
            }
            Thread.State state = thread.getState();
            if (state == Thread.State.TERMINATED) {
                return true;
            }
            return false;
        }

        @SneakyThrows
        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            this.stop();
            return true;
        }

        public void stop() {
            if (ObjectUtil.isNull(thread)) {
                return;
            }
            thread.interrupt();
            if (!thread.isInterrupted()) {
//                thread.stop();
            }
            thread = null;
        }
    }


    @Getter
    @Setter
    @AllArgsConstructor
    @NoArgsConstructor
    public static class TrackFutureInfo {
        private Future<?> future;

    }

}
